package com.activemqwrap.tools.tcpforward;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.advisory.DestinationSource;
import org.apache.activemq.command.ActiveMQQueue;
import org.jetbrains.annotations.NotNull;
import org.springframework.util.StringUtils;

import javax.jms.*;
import javax.jms.Queue;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.*;

public class ProxyMQ {

//    public  String[] socketArg = new String[]{"192.168.1.222", "9999", "192.168.1.230", "8081"};
//    public  String[] socketArg = new String[]{"192.168.1.222", "9999", "192.168.1.222", "8081"};
//    public  String DEFAULT_BROKER_HOST = "tcp://192.168.1.230:61616";
//    public  String userName = "system";
//    public  String password = "manager";
//    public  ActiveMQConnectionFactory connectionFactory;
//    public  String[] queueNames = new String[]{"input_Queue_sundata", "return_Queue_sundata"};

    public String[] socketArg;
    public String DEFAULT_BROKER_HOST;
    public String userName;
    public String password;
    public ActiveMQConnectionFactory connectionFactory;
    public String[] queueNames;

    public Hashtable<String, SocketExt> inSockets = new Hashtable<>();
    public Hashtable<String, SocketExt> outSockets = new Hashtable<>();
    public Hashtable<String, String> returnDataTable = new Hashtable<>();

    private ServerSocket rcvServerSocket;
    public int buffsize = 1024 * 128;
    public int buffsize_server = 1024 * 128;
    public int sleepTime = 10;
    public int checkReturnDataTable_sleepTime = 2;
    /**
     * 当前服务器ServerSocket的最大连接数
     */
    public final int MAX_CONNECTION_NUM = 5000;
    public boolean rcvRun = false;
    public boolean showDebug = false;

    public void start(int InOrOUt) {
        try {
            connectionFactory = new ActiveMQConnectionFactory(
                    userName,
                    password,
                    DEFAULT_BROKER_HOST
            );

            //1 数据输入
            if (InOrOUt == 1 || InOrOUt == 0) {
                String inIp = socketArg[0];
                int inPort = Integer.parseInt(socketArg[1]);
                rcvRun = true;
                rcvServerSocket = new ServerSocket(inPort, MAX_CONNECTION_NUM, InetAddress.getByName(inIp));
                System.out.println("开始监听：" + inIp + ":" + inPort);
                new Thread(() -> {
                    while (rcvRun) {
                        try {
                            // 用户连接到当前服务器的socket
                            Socket rcvSocket = rcvServerSocket.accept();
                            System.out.println(System.currentTimeMillis() + "-监听到新接入连接：" + rcvSocket.getRemoteSocketAddress());
                            new Thread(() -> {
                                rcvConnectThread(rcvSocket);
                            }).start();

                        } catch (Exception e) {
                            System.out.println("error-start-监听到新接入连接");
                        }
                    }
                }).start();

            }
            //2 数据转发
            if (InOrOUt == 2 || InOrOUt == 0) {
                new Thread(() -> {
                    try {
                        readMQ_app();
                    } catch (Exception e) {
                        System.out.println("error:readMQ_app---" + e.getMessage());
                    }
                }).start();

                new Thread(() -> {
                    try {
                        readMQ_server();
                    } catch (Exception e) {
                        System.out.println("error:readMQ_server---" + e.getMessage());
                    }
                }).start();

            }

        } catch (Exception e) {
            System.out.println("error:start---" + e.getMessage());
        }

    }

    public void stop() {
        try {
            rcvRun = false;
            rcvServerSocket.close();
        } catch (Exception e) {
            System.out.println("error:stop---" + e.getMessage());
        }

    }

    //处理新传入连接 接收数据写入队列
    public void rcvConnectThread(Socket rcvSocket) {
        try {
            String socketInfo = rcvSocket.getRemoteSocketAddress().toString();
            InputStream inStream = rcvSocket.getInputStream();
            OutputStream outStream = rcvSocket.getOutputStream();

            //记录新接入socket
            SocketExt socketExt = new SocketExt(rcvSocket, inStream, outStream);
            inSockets.put(socketExt.socketName, socketExt);

            //等待中的传入连接 循环查询hashtable是否已经返回并处理
            new Thread(() -> {
                while (!rcvSocket.isClosed()) {
                    try {
                        String retMsg = returnDataTable.get(socketInfo);
                        if (!StringUtils.isEmpty(retMsg)) {
                            byte[] msg = Base64.getDecoder().decode(retMsg);
                            if (!rcvSocket.isClosed()) {
                                outStream.write(msg, 0, msg.length);
                            }
                            returnDataTable.remove(socketInfo);
                        } else {
                            Thread.sleep(checkReturnDataTable_sleepTime);
                        }
                    } catch (Exception e) {
                        closeSocket(rcvSocket);
                        System.out.println("error:rcvConnectThread---outStream---" + e.getMessage());
                    }
                }
            }).start();

            //app传入数据 转送消息队列
            byte[] buff = new byte[buffsize];

            try {
                while (!rcvSocket.isClosed()) {

////////////////////方法3////////////////////////////
                    int allCount = 0;
                    int zeroRec = 0;
                    for (int a = 0; a < 50; a++) {
                        int inStackCount = inStream.available();
                        if (inStackCount == 65536) {
                            Rcv_Big_File(inStream, rcvSocket, socketInfo);
                            break;
                        }
                        if (inStackCount == allCount) {
                            zeroRec++;
                            if (zeroRec > 10) {
                                break;
                            }
                        } else {
                            allCount = inStackCount;
                        }
                        Thread.sleep(2);
                    }

                    if (allCount > 0) {
                        int size = inStream.read(buff);
                        if (size == -1) {
                            closeSocket(rcvSocket);
                        }
                        byte[] buff_tmp = Arrays.copyOf(buff, size);

                        String Str_Base64 = Base64.getEncoder().encodeToString(buff_tmp);
                        Str_Base64 = socketInfo + "@" + Str_Base64;
                        writeMQ(queueNames[0], Str_Base64);
                        System.out.println(System.currentTimeMillis() + "-" + socketInfo + "-客户输入数据转入队列：" + queueNames[0] + "-" + size);
                    }

                    Thread.sleep(sleepTime);
                }
            } catch (Exception e) {
                closeSocket(rcvSocket);
                System.out.println("error:rcvConnectThread---inStream---" + e.getMessage());
            }

        } catch (Exception e) {
            closeSocket(rcvSocket);
            System.out.println("error:rcvConnectThread---" + e.getMessage());
        }

    }

    //大文件特殊处理
    public void Rcv_Big_File(InputStream inputStream, Socket rcvSocket, String socketInfo) throws Exception {
        int allCount = 0;
        int zeroRec = 0;
        byte[] buff = new byte[buffsize];
        byte[] buff_rcv = new byte[]{};
        for (int a = 0; a < 100; a++) {
            int inStackCount = inputStream.available();
            if (inStackCount == 0) {
                zeroRec++;
                if (zeroRec > 10) {
                    break;
                }
            } else {
                int size = inputStream.read(buff);
                allCount = allCount + size;
                if (size == -1) {
                    closeSocket(rcvSocket);
                }
                byte[] buff_tmp = new byte[buff_rcv.length + size];
                System.arraycopy(buff_rcv, 0, buff_tmp, 0, buff_rcv.length);
                System.arraycopy(buff, 0, buff_tmp, buff_rcv.length, size);
                buff_rcv = buff_tmp;

            }
            Thread.sleep(2);
        }
        String Str_Base64 = Base64.getEncoder().encodeToString(buff_rcv);
        Str_Base64 = socketInfo + "@" + Str_Base64;
        writeMQ(queueNames[0], Str_Base64);
        System.out.println(System.currentTimeMillis() + "-" + socketInfo + "-客户输入-大文件-转入队列：" + queueNames[0] + "-" + buff_rcv.length);

    }

    //写消息队列
    public void writeMQ(String queueName, String msg) throws Exception {
        Connection connection = connectionFactory.createConnection();
        // 启动
        connection.start();
        // 创建会话 session，参数第一个是事务，第二个是签收
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建目的地，queue 或者 topic
        Queue queue = session.createQueue(queueName);
        // 创建消息的生产者
        MessageProducer producer = session.createProducer(queue);
        // 创建消息
        TextMessage textMessage = session.createTextMessage(msg);
        // 发送消息给 mq
        producer.send(textMessage);

        // 关闭
        session.close();
        connection.close();
//        System.out.println("队列消息发送成功~" + queueName);

    }

    //app端读取队列 转换数据并写入hashtable
    public void readMQ_app() throws Exception {
        String queueName = queueNames[1];
        Connection connection = connectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue(queueName);
        // 创建消费者
        MessageConsumer consumer = session.createConsumer(queue);
        while (true) {
            // 同步阻塞方式使用 receive()，超时之前一直等待
            // receive() 方法不带参数会一直等待
            // receive(Long timeout) 会等待指定时间后退出等待
            TextMessage receive = (TextMessage) consumer.receive();
            if (receive != null) {
//                System.out.println("接收到消息：" + receive.getText());
                String StrMsg = receive.getText();
                String[] retArray = StrMsg.split("@");
                returnDataTable.put(retArray[0], retArray[1]);

            } else {
                break;
            }
//            Thread.sleep(sleepTime);
        }
        consumer.close();
        session.close();
        connection.close();

    }

    //服务端读取队列
    public void readMQ_server() throws Exception {
        String queueName = queueNames[0];
        Connection connection = connectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue(queueName);
        // 创建消费者
        MessageConsumer consumer = session.createConsumer(queue);
        while (true) {
            // 同步阻塞方式使用 receive()，超时之前一直等待
            // receive() 方法不带参数会一直等待
            // receive(Long timeout) 会等待指定时间后退出等待
            TextMessage receive = (TextMessage) consumer.receive();
            if (receive != null) {
//                System.out.println("接收到消息：" + receive.getText());
                String StrMsg = receive.getText();
                //队列消息转发到后台
                server_trans_thread(StrMsg);

            } else {
                break;
            }
//            Thread.sleep(sleepTime);
        }
        consumer.close();
        session.close();
        connection.close();

    }

    //消息队列 服务器端转发接收 线程
    public void server_trans_thread(@NotNull String StrMsg) {
        String outIp = socketArg[2];
        int outPort = Integer.parseInt(socketArg[3]);
        String[] ID_Data = StrMsg.split("@");
        new Thread(() -> {
            try {

//                //判断是否有活动的socket
//                Socket serverSocket;
//                InputStream inputStream;
//                OutputStream outputStream;
//                SocketExt socketExtExsist = outSockets.get(ID_Data[0]);
//                if (!StringUtils.isEmpty(socketExtExsist)
//                        && !socketExtExsist.socket.isClosed()
//                        && !socketExtExsist.socket.isInputShutdown()
//                        && !socketExtExsist.socket.isOutputShutdown()) {
//                    serverSocket = socketExtExsist.socket;
//                    inputStream = socketExtExsist.inputStream;
//                    outputStream = socketExtExsist.outputStream;
//                } else {
//                    serverSocket = new Socket(outIp, outPort);
//                    //记录新转发socket
//                    inputStream = serverSocket.getInputStream();
//                    outputStream = serverSocket.getOutputStream();
//                    SocketExt socketExt = new SocketExt(serverSocket, inputStream, outputStream);
//                    outSockets.put(ID_Data[0], socketExt);
//                }

                //不判断是否有活动的socket 直接新开socket
                Socket serverSocket = new Socket(outIp, outPort);
                InputStream inputStream = serverSocket.getInputStream();
                OutputStream outputStream = serverSocket.getOutputStream();
                //记录新转发socket
                SocketExt socketExt = new SocketExt(serverSocket, inputStream, outputStream);
                outSockets.put(ID_Data[0], socketExt);

                //处理服务器返回数据
                byte[] buff = new byte[buffsize_server];
                new Thread(() -> {
                    try {
                        while (!serverSocket.isClosed()) {
                            int size = inputStream.read(buff);
                            if (size > -1) {
                                byte[] buff_s;
                                if (size < buffsize_server) {
                                    buff_s = Arrays.copyOf(buff, size);
                                } else {
                                    buff_s = buff;
                                }
                                String msgBase64 = Base64.getEncoder().encodeToString(buff_s);
                                msgBase64 = ID_Data[0] + "@" + msgBase64;
                                writeMQ(queueNames[1], msgBase64);
                                if (showDebug)
                                    System.out.println(System.currentTimeMillis() + "-" + ID_Data[0] + "-server返回数据转入队列:" + queueNames[1] + "-" + size);

                            } else {
                                System.out.println("-1:server_tran_thread---inputStream---closeSocket---");
                                closeSocket(serverSocket);
                            }
                            Thread.sleep(sleepTime);

                        }

                    } catch (Exception e) {
                        System.out.println("error:server_tran_thread---inputStream---" + e.getMessage());
                        closeSocket(serverSocket);
                    }

                }).start();

                //转发消息队列到服务器
                try {
                    byte[] msg = Base64.getDecoder().decode(ID_Data[1]);
                    if (!serverSocket.isClosed()) {
                        outputStream.write(msg, 0, msg.length);
                        System.out.println(System.currentTimeMillis() + "-" + ID_Data[0] + "-用户队列数据-转发server" + "-" + msg.length);
                    }
                } catch (Exception e) {
                    System.out.println("error:server_tran_thread---outputStream---" + e.getMessage());
                    closeSocket(serverSocket);
                }

            } catch (Exception e) {
                System.out.println("error:server_tran_thread---" + e.getMessage());
            }

        }).start();
    }

    public void closeSocket(Socket socket) {
        try {
            //预留关闭socket
            System.out.println(System.currentTimeMillis() + "-准备关闭socket" + "-" + socket);
            socket.shutdownInput();
            socket.shutdownOutput();
            socket.close();
        } catch (Exception e) {
            System.out.println("error:closeSocket---" + e.getMessage());
        }

    }

    //清除所有queue
    public void getAllQueueAndDelete() {
        int stopFlag = 0;

        while (stopFlag < 5) {
            try {
                ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                        userName,
                        password,
                        DEFAULT_BROKER_HOST
                );
                ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
                connection.start();
                DestinationSource ds = connection.getDestinationSource();
                Set<ActiveMQQueue> queues = ds.getQueues();

                for (ActiveMQQueue activeMQQueue : queues) {
                    try {
                        String queueName = activeMQQueue.getQueueName();
                        ActiveMQConnection conn = (ActiveMQConnection) connectionFactory.createConnection();
                        conn.destroyDestination(new ActiveMQQueue(queueName));
//                    System.out.println("Deleted---" + activeMQQueue.getQueueName());
                    } catch (Exception e) {
                        System.out.println("getAllQueueAndDelete" + e.getMessage());
                    }
                }
                if (queues.size() == 0) {
                    stopFlag++;
                }
                System.out.println("deleted--" + queues.size());
                connection.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }

}
